package cn.vendsystem.com.consumer;


import cn.vendcommon.com.entity.constants.MQConstant;
import cn.vendsystem.com.service.InventoryService;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.HashMap;
import java.util.Map;


@Slf4j
@Component
public class InventoryConsumer {

    @Autowired
    private InventoryService inventoryService;
    @Autowired
    private RabbitTemplate rabbitTemplate;

//    @RabbitListener(queues = MQConstant.INVENTORY_QUEUE, containerFactory = "rabbitListenerContainerFactory")
//    public void onMessage(Map<String, Object> msg) {
//        Integer machineId = (Integer) msg.get("machineId");
//        Map<String, Integer> intelligentMap = (Map<String, Integer>) msg.get("intelligentMap");
//        inventoryService.deductionInventoryByMachineId(machineId, intelligentMap);
//    }

    /**
     * 扣减库存消息监听
     * 手动 ACK + 异常处理，异常消息进入死信队列
     * 消息到达 -> 校验 -> 执行扣减库存 -> 成功 ACK
     *                              -> 异常 Reject -> 死信队列 -> 回滚队列( TODO 逻辑暂未实现)
     * @param msg  消费的消息体，已通过 Jackson2JsonMessageConverter 转成 Map
     * @param channel  RabbitMQ 的底层通道对象，用于手动确认或拒绝消息
     * @param tag  消息的唯一标识，用于 basicAck 或 basicReject
     */
    @RabbitListener(queues = MQConstant.INVENTORY_QUEUE, containerFactory = "rabbitListenerContainerFactory")
    public void onMessage(Map<String, Object> msg, Channel channel, @Header(AmqpHeaders.DELIVERY_TAG) long tag) {
        try {
            // 解析消息
            Integer machineId = (Integer) msg.get("machineId");
            if (machineId == null) {
                throw new IllegalArgumentException("machineId 不能为空");
            }

            Map<String, Integer> intelligentMap = (Map<String, Integer>) msg.get("intelligentMap");
            if (intelligentMap == null || intelligentMap.isEmpty()) {
                throw new IllegalArgumentException("intelligentMap 不能为空");
            }

            // 执行扣减库存逻辑
            inventoryService.deductionInventoryByMachineId(machineId, intelligentMap);

            // <优化> 消息处理成功，手动 ACK
            channel.basicAck(tag, false);  // 告诉 RabbitMQ 该消息消费成功
            log.info("库存扣减成功，machineId={}", machineId);

        } catch (Exception e) {
            log.error("库存扣减失败，消息发送到死信队列，msg={}", msg, e);
            try {
                // false拒绝消息，不重新入队且消息会进入死信队列
                channel.basicReject(tag, false);
            } catch (IOException ioException) {
                log.error("消息拒绝失败", ioException);
            }
        }
    }

    /**
     * 死信队列消费者
     * 由于扣减库存失败，再次尝试重新扣减库存
     * @param msg
     * @param channel
     * @param tag
     * @throws IOException
     */
    @RabbitListener(queues = MQConstant.INVENTORY_DLQ)
    public void handleDeadLetter(Map<String, Object> msg, Channel channel,
                                 @Header(AmqpHeaders.DELIVERY_TAG) long tag) throws IOException {
        try {
            Integer machineId = (Integer) msg.get("machineId");
            Map<String, Integer> intelligentMap = (Map<String, Integer>) msg.get("intelligentMap");

            // 可以选择再次尝试扣减，或者做报警/日志
            inventoryService.deductionInventoryByMachineId(machineId, intelligentMap);

            channel.basicAck(tag, false);
            log.info("死信队列处理成功，machineId={}", machineId);
        } catch (Exception e) {
            log.error("死信队列处理失败，msg={}", msg, e);
//            channel.basicReject(tag, false); // 再次进入死信队列或手动处理

            Map<String, Object> rollbackMsg = new HashMap<>();
            rollbackMsg.put("machineId", msg.get("machineId"));
            rollbackMsg.put("intelligentMap", msg.get("intelligentMap"));
            rollbackMsg.put("xid", msg.get("xid"));
            rollbackMsg.put("reason", e.getMessage());
            // TODO 回滚队列逻辑暂未实现
            rabbitTemplate.convertAndSend(MQConstant.ROLLBACK_EXCHANGE, MQConstant.ROLLBACK_ROUTING_KEY, rollbackMsg);
            channel.basicAck(tag, false);
        }
    }
}
